package org.ghost.springboot2.demo.service.impl;

import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.ghost.springboot2.demo.producer.DefaultProducer;
import org.ghost.springboot2.demo.service.IDefaultProducerService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.util.Assert;

import java.nio.charset.StandardCharsets;

@Service
public class DefaultProducerServiceImpl implements IDefaultProducerService {
    @Autowired
    private DefaultProducer defaultProducer;

    @Override
    public SendResult syncSend(String topic, String tags, String msg) throws InterruptedException, RemotingException, MQClientException, MQBrokerException {
        Assert.hasText(topic, "topic can not blank");
        Assert.hasText(msg, "msg can not blank");

        Message message = new Message(topic, tags, msg.getBytes(StandardCharsets.UTF_8));
        return this.defaultProducer.getDefaultProducer().send(message);
    }

    @Override
    public SendResult syncSend(String topic, String tags, String msg, long timeout) throws InterruptedException, RemotingException, MQClientException, MQBrokerException {
        Assert.hasText(topic, "topic can not blank");
        Assert.hasText(msg, "msg can not blank");

        Message message = new Message(topic, tags, msg.getBytes(StandardCharsets.UTF_8));
        return this.defaultProducer.getDefaultProducer().send(message, timeout);
    }

    @Override
    public void asyncSend(String topic, String tags, String msg, SendCallback callback) throws RemotingException, MQClientException, InterruptedException {
        Assert.hasText(topic, "topic can not blank");
        Assert.hasText(msg, "msg can not blank");
        Assert.notNull(callback, "callback can not null");

        Message message = new Message(topic, tags, msg.getBytes(StandardCharsets.UTF_8));
        this.defaultProducer.getDefaultProducer().send(message, callback);
    }

    @Override
    public void asyncSend(String topic, String tags, String msg, SendCallback callback, long timeout) throws RemotingException, MQClientException, InterruptedException {
        Assert.hasText(topic, "topic can not blank");
        Assert.hasText(msg, "msg can not blank");
        Assert.notNull(callback, "callback can not null");

        Message message = new Message(topic, tags, msg.getBytes(StandardCharsets.UTF_8));
        this.defaultProducer.getDefaultProducer().send(message, callback, timeout);
    }

    @Override
    public void sendOneWay(String topic, String tags, String msg) throws RemotingException, MQClientException, InterruptedException {
        Assert.hasText(topic, "topic can not blank");
        Assert.hasText(msg, "msg can not blank");

        Message message = new Message(topic, tags, msg.getBytes(StandardCharsets.UTF_8));
        this.defaultProducer.getDefaultProducer().sendOneway(message);
    }

    @Override
    public SendResult syncSend(String topic, String tags, String msg, MessageQueueSelector selector, Object arg) throws InterruptedException, RemotingException, MQClientException, MQBrokerException {
        Assert.hasText(topic, "topic can not blank");
        Assert.hasText(msg, "msg can not blank");
        Assert.notNull(selector, "selector can not null");

        Message message = new Message(topic, tags, msg.getBytes(StandardCharsets.UTF_8));
        return this.defaultProducer.getDefaultProducer().send(message, selector, arg);
    }

    @Override
    public SendResult syncSend(String topic, String tags, String msg, MessageQueueSelector selector, Object arg, long timeout) throws InterruptedException, RemotingException, MQClientException, MQBrokerException {
        Assert.hasText(topic, "topic can not blank");
        Assert.hasText(msg, "msg can not blank");
        Assert.notNull(selector, "selector can not null");

        Message message = new Message(topic, tags, msg.getBytes(StandardCharsets.UTF_8));
        return this.defaultProducer.getDefaultProducer().send(message, selector, arg, timeout);
    }

    @Override
    public void asyncSend(String topic, String tags, String msg, MessageQueueSelector selector, Object arg, SendCallback callback) throws RemotingException, MQClientException, InterruptedException {
        Assert.hasText(topic, "topic can not blank");
        Assert.hasText(msg, "msg can not blank");
        Assert.notNull(selector, "selector can not null");
        Assert.notNull(callback, "callback can not null");

        Message message = new Message(topic, tags, msg.getBytes(StandardCharsets.UTF_8));
        this.defaultProducer.getDefaultProducer().send(message, selector, arg, callback);
    }

    @Override
    public void asyncSend(String topic, String tags, String msg, MessageQueueSelector selector, Object arg, SendCallback callback, long timeout) throws RemotingException, MQClientException, InterruptedException {
        Assert.hasText(topic, "topic can not blank");
        Assert.hasText(msg, "msg can not blank");
        Assert.notNull(selector, "selector can not null");
        Assert.notNull(callback, "callback can not null");

        Message message = new Message(topic, tags, msg.getBytes(StandardCharsets.UTF_8));
        this.defaultProducer.getDefaultProducer().send(message, selector, arg, callback, timeout);
    }

//    @Scheduled(fixedDelay = 5 * 1000)
//    public void sendMessage() throws InterruptedException, RemotingException, MQClientException, MQBrokerException {
//        Message message = new Message("topic", "tag1", ("hello " + ChineseNameUtil.getChineseName()).getBytes());
//        //发送
//        SendResult sendResult = defaultProducer.getDefaultProducer().send(message);
//    }
}
